package com.spi.controller;

import com.alibaba.fastjson.JSON;
import com.spi.config.MqTransactionListener;
import domain.User;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.client.producer.TransactionMQProducer;
import org.apache.rocketmq.client.producer.TransactionSendResult;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.apache.rocketmq.spring.support.RocketMQHeaders;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import java.time.LocalDateTime;
import java.util.Date;
import java.util.HashMap;
import java.util.Map;
import java.util.Random;

@RequestMapping("/user")
@RestController
@Slf4j
public class UserController {

    @Autowired
    private RocketMQTemplate rocketMQTemplate ;

    @Autowired
    private MqTransactionListener mqTransactionListener ;


    @RequestMapping("/sendMessageMQ")
    public void test00(){
        System.out.println("准备发送消息");
        rocketMQTemplate.asyncSend("test-topic", "生产者发送消息。。。", new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {
                System.out.println("消息投递成功。。");
            }

            @Override
            public void onException(Throwable throwable) {
                System.out.println("消息投递失败。。");
            }
        });
    }

    @RequestMapping("/sendMessageMQ1")
    public void test01(){
        System.out.println("准备发送消息consumer-topic1");
        rocketMQTemplate.asyncSend("consumer-topic1", "生产者发送消息。。consumer-topic1。", new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {
                System.out.println("consumer-topic1消息投递成功。。");
            }

            @Override
            public void onException(Throwable throwable) {
                System.out.println("consumer-topic1消息投递失败。。");
            }
        });
    }

    /**
     * 异步发送
     */
    @RequestMapping("/sendMessageMQ2")
    public void test02(){
        System.out.println("准备发送消息consumer-topic2");
        Map<String,String> map = new HashMap<>();
        map.put("content","生产者发送消息。。consumer-topic2.");
        //获取时间 年月日时分秒
        LocalDateTime now = LocalDateTime.now();
        String year = String.valueOf(now.getYear());
        String month = String.valueOf(now.getMonthValue());
        String day = String.valueOf(now.getDayOfMonth());
        String hour = String.valueOf(now.getHour());
        String minute = String.valueOf(now.getMinute());
        String second = String.valueOf(now.getSecond());
        String date = year + month + day + hour + minute + second;
        map.put("msgid",date);
        Message<Map<String,String>> build = MessageBuilder.withPayload(map).build();
        rocketMQTemplate.asyncSend("consumer-topic2", build, new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {
                String msgId = sendResult.getMsgId();
                System.out.println("consumer-topic2消息投递成功。。消息id为：" + msgId);
            }

            @Override
            public void onException(Throwable throwable) {
                System.out.println("consumer-topic2消息投递失败。。");
            }
        });
    }

    /**
     * 有序同步发送
     */
    @RequestMapping("/sendMessageMQ22")
    public void test022(){
        String hashKey = "globalOrderlyKey"; //全局顺序标识
        for (int i = 0; i < 5; i++) {
            //System.out.println("准备发送消息consumer-topic2");
            Map<String,String> map = new HashMap<>();
            map.put("content","第" + i + "生产者发送消息。。consumer-topic2.");
            //获取时间 年月日时分秒
            LocalDateTime now = LocalDateTime.now();
            String year = String.valueOf(now.getYear());
            String month = String.valueOf(now.getMonthValue());
            String day = String.valueOf(now.getDayOfMonth());
            String hour = String.valueOf(now.getHour());
            String minute = String.valueOf(now.getMinute());
            String second = String.valueOf(now.getSecond());
            String date = year + month + day + hour + minute + second + i;
            map.put("msgid",date);
            Message<Map<String,String>> build = MessageBuilder.withPayload(map).build();
            SendResult result = rocketMQTemplate.syncSendOrderly("consumer-topic2", build, hashKey);
            System.out.println("第" + i + "consumer-topic2消息投递成功。。消息id为：" + result.getMsgId() + "发送到队列：" + result.getMessageQueue());
        }

    }
    @RequestMapping("/sendMessageMQ3")
    public void test03(){
        System.out.println("准备发送消息consumer-topic3");
        Message<String> build = MessageBuilder.
                withPayload("生产者发送消息。。consumer-topic3.").
                setHeader("tag", "tag5").
                build();
        rocketMQTemplate.asyncSend("consumer-topic3:tag5",build, new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {
                System.out.println("consumer-topic3消息投递成功。。消息msgid为：" + sendResult.getMsgId());
                System.out.println("consumer-topic3消息投递成功。。消息offsetmsgid为：" + sendResult.getOffsetMsgId());
            }

            @Override
            public void onException(Throwable throwable) {
                System.out.println("consumer-topic3消息投递失败。。");
            }
        });
    }
    @RequestMapping("/sendMessageMQ33")
    public void test033(){
        System.out.println("准备发送消息consumer-topic33");
        Message<String> build = MessageBuilder.
                withPayload("生产者发送消息。。consumer-topic33.").
                setHeader(RocketMQHeaders.TAGS, "tag333").
                        build();
        rocketMQTemplate.asyncSend("consumer-topic3",build, new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {
                System.out.println("consumer-topic33消息投递成功。。消息msgid为：" + sendResult.getMsgId());
                System.out.println("consumer-topic33消息投递成功。。消息offsetmsgid为：" + sendResult.getOffsetMsgId());
            }

            @Override
            public void onException(Throwable throwable) {
                System.out.println("consumer-topic33消息投递失败。。");
            }
        });
    }

    /**
     * 发送事务消息
     */
    @RequestMapping("/sendMessageMQ4")
    public void test04(){
        User user = new User();
        user.setId(1);
        user.setAge("33");
        user.setName("小王");
        user.setHobby("羽毛球");
        String msg = JSON.toJSONString(user);
        Message<String> build = MessageBuilder.withPayload(msg).build();
        ((TransactionMQProducer) rocketMQTemplate.getProducer()).setTransactionListener(mqTransactionListener);
        log.info("producer准备发送数据。。");
        TransactionSendResult transactionSendResult = rocketMQTemplate.sendMessageInTransaction("consumer-topic4", build, null);
        log.info("事务消息结果：{}", transactionSendResult);
    }
}
